//! Elasticsearch数据源实现
//! 
//! 基于HTTP API的Elasticsearch数据源插件

use async_trait::async_trait;
use serde::{Serialize, Deserialize};
use serde_json::Value;
use std::collections::HashMap;
use std::time::Instant;
use reqwest::Client;

use crate::{DataSource, DataSourceQuery, DataSourceResult, DataSourceStats};

/// Elasticsearch数据源配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ElasticsearchConfig {
    /// Elasticsearch节点URL
    pub url: String,
    /// 用户名（可选）
    pub username: Option<String>,
    /// 密码（可选）
    pub password: Option<String>,
    /// 默认索引
    pub default_index: String,
    /// 连接超时时间（秒）
    pub connect_timeout: u64,
    /// 查询超时时间（秒）
    pub query_timeout: u64,
}

impl Default for ElasticsearchConfig {
    fn default() -> Self {
        Self {
            url: "http://localhost:9200".to_string(),
            username: None,
            password: None,
            default_index: "logs".to_string(),
            connect_timeout: 30,
            query_timeout: 60,
        }
    }
}

/// Elasticsearch数据源
pub struct ElasticsearchDataSource {
    config: ElasticsearchConfig,
    client: Option<Client>,
}

impl ElasticsearchDataSource {
    /// 创建新的Elasticsearch数据源
    pub fn new(config: ElasticsearchConfig) -> Self {
        Self {
            config,
            client: None,
        }
    }

    /// 获取HTTP客户端
    fn get_client(&self) -> anyhow::Result<&Client> {
        self.client.as_ref().ok_or_else(|| {
            anyhow::anyhow!("Elasticsearch数据源未连接")
        })
    }

    /// 将SQL查询转换为Elasticsearch查询
    fn sql_to_es_query(&self, sql: &str) -> anyhow::Result<Value> {
        // 简化的SQL到ES查询转换
        let sql_lower = sql.to_lowercase();
        
        if sql_lower.contains("select * from") {
            // 简单的全表查询
            Ok(serde_json::json!({
                "query": {
                    "match_all": {}
                },
                "size": 100
            }))
        } else if sql_lower.contains("where") {
            // 带条件的查询
            // 这里是简化实现，实际应该解析SQL语法
            Ok(serde_json::json!({
                "query": {
                    "query_string": {
                        "query": "*"
                    }
                },
                "size": 100
            }))
        } else {
            // 默认查询
            Ok(serde_json::json!({
                "query": {
                    "match_all": {}
                },
                "size": 10
            }))
        }
    }

    /// 执行Elasticsearch查询
    async fn execute_es_query(&self, index: &str, query: &Value) -> anyhow::Result<Value> {
        let client = self.get_client()?;
        let url = format!("{}/{}/_search", self.config.url, index);

        let mut request = client.post(&url).json(query);

        // 添加认证信息
        if let (Some(username), Some(password)) = (&self.config.username, &self.config.password) {
            request = request.basic_auth(username, Some(password));
        }

        let response = request.send().await?;
        
        if response.status().is_success() {
            let result: Value = response.json().await?;
            Ok(result)
        } else {
            let error_text = response.text().await?;
            Err(anyhow::anyhow!("Elasticsearch查询失败: {}", error_text))
        }
    }

    /// 转换ES结果为标准格式
    fn transform_es_result(&self, es_result: Value) -> anyhow::Result<Vec<Value>> {
        let hits = es_result
            .get("hits")
            .and_then(|h| h.get("hits"))
            .and_then(|h| h.as_array())
            .ok_or_else(|| anyhow::anyhow!("无效的Elasticsearch响应格式"))?;

        let mut results = Vec::new();
        for hit in hits {
            if let Some(source) = hit.get("_source") {
                results.push(source.clone());
            }
        }

        Ok(results)
    }
}

#[async_trait]
impl DataSource for ElasticsearchDataSource {
    fn name(&self) -> &str {
        "Elasticsearch"
    }
    
    fn description(&self) -> &str {
        "Elasticsearch搜索引擎数据源"
    }
    
    async fn connect(&mut self) -> anyhow::Result<()> {
        let client = Client::builder()
            .timeout(std::time::Duration::from_secs(self.config.connect_timeout))
            .build()?;

        // 测试连接
        let url = format!("{}/_cluster/health", self.config.url);
        let mut request = client.get(&url);

        if let (Some(username), Some(password)) = (&self.config.username, &self.config.password) {
            request = request.basic_auth(username, Some(password));
        }

        let response = request.send().await?;
        
        if !response.status().is_success() {
            return Err(anyhow::anyhow!("连接Elasticsearch失败: {}", response.status()));
        }

        self.client = Some(client);
        Ok(())
    }
    
    async fn disconnect(&mut self) -> anyhow::Result<()> {
        self.client = None;
        Ok(())
    }
    
    async fn is_connected(&self) -> bool {
        self.client.is_some()
    }
    
    async fn execute_query(&mut self, query: DataSourceQuery) -> anyhow::Result<DataSourceResult> {
        let start_time = Instant::now();
        
        // 转换SQL查询为ES查询
        let es_query = match self.sql_to_es_query(&query.content) {
            Ok(q) => q,
            Err(e) => {
                let duration = start_time.elapsed();
                return Ok(DataSourceResult {
                    success: false,
                    data: Value::Null,
                    error: Some(format!("SQL转换失败: {}", e)),
                    stats: DataSourceStats {
                        execution_time_ms: duration.as_millis() as f64,
                        rows_processed: 0,
                        bytes_processed: 0,
                    },
                });
            }
        };

        // 执行查询
        match self.execute_es_query(&self.config.default_index, &es_query).await {
            Ok(es_result) => {
                let duration = start_time.elapsed();
                
                // 转换结果
                match self.transform_es_result(es_result) {
                    Ok(results) => {
                        Ok(DataSourceResult {
                            success: true,
                            data: Value::Array(results.clone()),
                            error: None,
                            stats: DataSourceStats {
                                execution_time_ms: duration.as_millis() as f64,
                                rows_processed: results.len() as u64,
                                bytes_processed: 0, // TODO: 计算实际字节数
                            },
                        })
                    }
                    Err(e) => {
                        Ok(DataSourceResult {
                            success: false,
                            data: Value::Null,
                            error: Some(format!("结果转换失败: {}", e)),
                            stats: DataSourceStats {
                                execution_time_ms: duration.as_millis() as f64,
                                rows_processed: 0,
                                bytes_processed: 0,
                            },
                        })
                    }
                }
            }
            Err(e) => {
                let duration = start_time.elapsed();
                
                Ok(DataSourceResult {
                    success: false,
                    data: Value::Null,
                    error: Some(e.to_string()),
                    stats: DataSourceStats {
                        execution_time_ms: duration.as_millis() as f64,
                        rows_processed: 0,
                        bytes_processed: 0,
                    },
                })
            }
        }
    }
    
    async fn health_check(&self) -> anyhow::Result<Value> {
        if !self.is_connected().await {
            return Ok(serde_json::json!({
                "status": "unhealthy",
                "database": "Elasticsearch",
                "connected": false,
                "url": self.config.url
            }));
        }

        // 检查集群健康状态
        let client = self.get_client()?;
        let url = format!("{}/_cluster/health", self.config.url);
        let mut request = client.get(&url);

        if let (Some(username), Some(password)) = (&self.config.username, &self.config.password) {
            request = request.basic_auth(username, Some(password));
        }

        match request.send().await {
            Ok(response) => {
                if response.status().is_success() {
                    let health: Value = response.json().await?;
                    Ok(serde_json::json!({
                        "status": "healthy",
                        "database": "Elasticsearch",
                        "connected": true,
                        "url": self.config.url,
                        "cluster_health": health
                    }))
                } else {
                    Ok(serde_json::json!({
                        "status": "unhealthy",
                        "database": "Elasticsearch",
                        "connected": true,
                        "url": self.config.url,
                        "error": format!("健康检查失败: {}", response.status())
                    }))
                }
            }
            Err(e) => {
                Ok(serde_json::json!({
                    "status": "unhealthy",
                    "database": "Elasticsearch",
                    "connected": true,
                    "url": self.config.url,
                    "error": e.to_string()
                }))
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn create_test_config() -> ElasticsearchConfig {
        ElasticsearchConfig {
            url: "http://localhost:9200".to_string(),
            username: None,
            password: None,
            default_index: "test_index".to_string(),
            connect_timeout: 5,
            query_timeout: 30,
        }
    }

    #[test]
    fn test_elasticsearch_config_default() {
        let config = ElasticsearchConfig::default();
        assert_eq!(config.url, "http://localhost:9200");
        assert_eq!(config.default_index, "logs");
        assert_eq!(config.connect_timeout, 30);
        assert_eq!(config.query_timeout, 60);
    }

    #[tokio::test]
    async fn test_elasticsearch_data_source_creation() {
        let config = create_test_config();
        let es_source = ElasticsearchDataSource::new(config);

        assert_eq!(es_source.name(), "Elasticsearch");
        assert_eq!(es_source.description(), "Elasticsearch搜索引擎数据源");
        assert!(!es_source.is_connected().await);
    }

    #[test]
    fn test_sql_to_es_query_conversion() {
        let config = create_test_config();
        let es_source = ElasticsearchDataSource::new(config);

        // 测试全表查询
        let query1 = es_source.sql_to_es_query("SELECT * FROM logs").unwrap();
        assert!(query1.get("query").is_some());
        assert!(query1.get("size").is_some());

        // 测试带条件查询
        let query2 = es_source.sql_to_es_query("SELECT * FROM logs WHERE level = 'ERROR'").unwrap();
        assert!(query2.get("query").is_some());
    }

    #[tokio::test]
    async fn test_elasticsearch_health_check_disconnected() {
        let config = create_test_config();
        let es_source = ElasticsearchDataSource::new(config);
        
        let health = es_source.health_check().await.unwrap();
        assert_eq!(health["status"], "unhealthy");
        assert_eq!(health["connected"], false);
        assert_eq!(health["database"], "Elasticsearch");
    }

    #[tokio::test]
    async fn test_elasticsearch_query_without_connection() {
        let config = create_test_config();
        let mut es_source = ElasticsearchDataSource::new(config);
        
        let query = DataSourceQuery {
            id: "test".to_string(),
            content: "SELECT * FROM logs".to_string(),
            parameters: HashMap::new(),
        };
        
        let result = es_source.execute_query(query).await.unwrap();
        assert!(!result.success);
        assert!(result.error.is_some());
    }
}
